<?php
/*
 * @Descripttion: 基于 redis 实现的队列服务驱动
 * @version: 
 * @Author: ewgof
 * @Date: 2021-04-27 14:15:33
 * @LastEditors: ewgof
 * @LastEditTime: 2022-12-03 22:11:45
 */

namespace queue\drivers;

// require './interfaces/DriverInterface.php';
use queue\interfaces\DriverInterface;

class RedisDriver implements DriverInterface
{
    // 配置信息
    private static $config = [];
    // redis 句柄
    private static $redis = null;
    private static $errMsg = '';
    /**
     * @name:
     * @msg: 连接队列服务提供者，比如 redis、rabbitMQ 等等
     * @param {array} $config 队列任务提供者相关配置
     * @return {resource} 返回相关的连接资源句柄
     */
    public static function connect()
    {
        if (empty(self::$config)) {
            self::$config = require_once __DIR__ . '/config.php';
        }
        try {
            $redis = new \Redis();
            $redis_config = self::$config['redis'];
            // var_dump($redis_config);return ;

            $redis->pconnect(
                $redis_config['host'],
                $redis_config['port'],
                $redis_config['timeout']
            );
            // $redis->connect('127.0.0.1', 6379, 60);
            if (!empty($redis_config['passwd'])) {
                $redis->auth($redis_config['passwd']);
            }
            if (!empty($redis_config['dbname'])) {
                $redis->select($redis_config['dbname']);
            } else {
                $redis->select(0);
            }
            self::$redis = $redis;
            return true;
        } catch (\Exception $e) {
            self::$errMsg = $e->getMessage();
            return false;
        }
    }

    /**
     * @name:
     * @msg: 推送消息到队列服务提供者
     * @param {array} $handler 指明任务处理逻辑，该参数结构如下： ['class_name', 'method'],其中 class_name 表示类名， method 表示方法名
     * @param {array} $params 传递给方法的参数
     * @param {integer} $retry_times 队列任务执行错误重试次数，默认 0 ，无限重试
     * @param {string} $queue_name 队列名称
     * @param {array} $init_params 传递给方法的参数,如果不是静态方式访问，此参数将会传递到构造函数中，如果 $handler 中指定的 method 是一个静态方法时，此参数被忽略
     * @return {bool}
     */
    public static function push(array $handler, array $params = [], int $retry_times = 0, string $queue_name = 'message_queue_list', array $init_params = [])
    {
        $redis = self::$redis;
        $retried_times = !empty($params['retried_times']) ? $params['retried_times'] : 0;
        unset($params['retried_times']);
        $message_id = generateMessageId('rq_');
        $msg = json_encode(compact('handler', 'params', 'init_params', 'static', 'retry_times', 'retried_times', 'message_id'));
        $r = $redis->lPush($queue_name, $msg);
        return (bool)$r;
    }


    /**
     * @name:
     * @msg: 推送消息到队列服务提供者,推送到延时队列，如果使用 redis 作为队列服务提供者，则不支持延时队列
     * @param {array} $handler 指明任务处理逻辑，该参数结构如下： ['class_name', 'method'],其中 class_name 表示类名， method 表示方法名
     * @param {array} $params 传递给方法的参数
     * @param {integer} $retry_times 队列任务执行错误重试次数，默认 0 ，无限重试
     * @param {string} $queue_name 队列名称
     * @param {array} $init_params 传递给方法的参数,如果不是静态方式访问，此参数将会传递到构造函数中，$static = true 时，此参数被忽略
     * @param {integer} $delays 延时时间，单位 s
     * @return {bool}
     */
    public static function delayPush(array $handler, array $params = [], int $retry_times = 0, string $queue_name = 'message_queue_list', array $init_params = [], int $delays = 10)
    {
        return self::push($handler, $params, $retry_times, $queue_name, $init_params, $delays);
    }


    /**
     * @name:
     * @msg: 从队列服务提供者拿取消息
     * @param {mixed} $queue_name 队列名称,如果是监听多个队列，把名称放数组内
     * @return {string} json string
     */
    public static function pull($queue_name, int $time_out = 0)
    {
        if (is_array($queue_name)) {
            $queue_name[] = self::$config['retry_queue'];
        } else {
            $queue_name = [
                $queue_name,
                self::$config['retry_queue'],
            ];
        }
        $msg = self::$redis->brPop($queue_name, $time_out);
        if (empty($msg)) {
            return $msg;
        }
        if (empty($msg[1])) {
            $msg = null;
        } else {
            $msg = json_decode($msg[1], true);
        }
        return $msg;
    }


    /**
     * @name: 
     * @msg: 消息处理确认
     * @param {string} $queue_name 队列标识 / 队列名称
     * @param {array} $queue_msg_arr 消息体
     * @param {string} $result 处理结果，failed：失败；success：成功
     * @return {*}
     */
    public static function confirm(array $queue_msg_arr, string $result = 'success')
    {
        if ('failed' == $result) {
            if ($queue_msg_arr['retry_times'] == 0 || $queue_msg_arr['retried_times'] <= $queue_msg_arr['retry_times']) {
                // 任务执行失败重新入队列
                $queue_msg_arr['params']['retried_times'] = ++$queue_msg_arr['retried_times'];
                self::push($queue_msg_arr['handler'], $queue_msg_arr['params'], $queue_msg_arr['retry_times'], self::$config['retry_queue'], $queue_msg_arr['init_params']);
                $msg = '任务处理失败重新进入队列，重试次数：' . $queue_msg_arr['retried_times'] . '，message => ' . json_encode($queue_msg_arr);
                echo $msg . PHP_EOL;
                logw($msg, 'error');
            } elseif ($queue_msg_arr['retried_times'] > $queue_msg_arr['retry_times']) {
                self::push($queue_msg_arr['handler'], $queue_msg_arr['params'], $queue_msg_arr['retry_times'], self::$config['dead_letter_queue'], $queue_msg_arr['init_params']);
                $msg = '任务处理失败并且超过最大重试次数，进入死信队列。retried_times => ' . $queue_msg_arr['retried_times'] . '，message => ' . json_encode($queue_msg_arr, JSON_UNESCAPED_UNICODE);
                echo $msg . PHP_EOL;
                Log::record($msg, 'queueinfo');
            }
        }
        return true;
    }

    /**
     * @name:
     * @msg:
     * @param {string} $queue_name 队列名称
     * @param {bool} $frozen 是否在队列里面还有消息的时候强制删除队列
     * @return {bool}
     */
    public static function delete(string $queue_name, bool $frozen = false)
    {
        if (!$frozen) {
            $len = self::$redis->lLen($queue_name);
            if ($len) {
                self::$errMsg = '队列不为空不能删除';
                return false;
            }
        }
        self::$redis->del($queue_name);
        return true;
    }


        /**
     * @name: 
     * @msg: 获取每个队列消息数量
     * @return 
     */
    public static function info() {
        $queue_names = self::$config['queue_name'];
        $queue_names[] = self::$config['retry_queue'];

        $info = [];
        foreach ($queue_names as $name) {
            $info[$name] = self::$redis->lLen($name);
        }
        return $info;
    }
}
